package ex.datastream.windows.windowFunctions;

import ex.datastream.ApiFrame;
import ex.datastream.functions.richFunction.MyProcessWindowFunction;
import ex.vo.SensorReading;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;



/**
 * keyBy+window+reduce(ReduceFunction&ProcessWindowFunction)
 * 输入
 C:\Users\Administrator>nc -lp 9999
 id-01,1707483562,1
 id-01,1707483562,1
 id-01,1707483562,1
 * 输出
 start:2024-02-28 15:13:50;end:2024-02-28 15:13:52;key:id-01
 1> (1709104430000,SensorReading{sensorId='id-01', timeStamp=1707483562, temperature=1.0, lowOrhigt=null, status=null})
 start:2024-02-28 15:13:52;end:2024-02-28 15:13:54;key:id-01
 1> (1709104432000,SensorReading{sensorId='id-01', timeStamp=1707483562, temperature=1.0, lowOrhigt=null, status=null})
 start:2024-02-28 15:13:54;end:2024-02-28 15:13:56;key:id-01
 1> (1709104434000,SensorReading{sensorId='id-01', timeStamp=1707483562, temperature=1.0, lowOrhigt=null, status=null})
 */
public class ReduceFuncTest extends ApiFrame {
    public static void main(String[] args) throws Exception {
        ReduceFuncTest windows=new ReduceFuncTest();
        windows.getEnv();
        DataStream<SensorReading> dataStream =windows.getSensorReading();

        KeyedStream<SensorReading, String> keyedStream = dataStream.keyBy(value -> value.sensorId);

        WindowedStream<SensorReading, String,TimeWindow> windowedStream = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(2)));

        windowedStream.reduce(new MyReduceFunction(), new MyProcessWindowFunction()).print();

        windows.env.execute();
    }



    private static class MyReduceFunction implements ReduceFunction<SensorReading> {
        public SensorReading reduce(SensorReading r1, SensorReading r2) {
            return r1.temperature > r2.temperature ? r1 : r2;
        }
    }
}
